package com.vion.consumer;

import com.alibaba.fastjson.JSONObject;
import com.vion.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author vion
 **/
@Component
@Slf4j
public class TFlowConsumer {


    @KafkaListener(topicPartitions ={@TopicPartition(topic = Constants.KAFKA_TFLOW,partitions = {"0"})}, containerFactory = "ackContainerFactory")
    public void vehicleData(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record : recordList) {
            JSONObject jsonObject = JSONObject.parseObject(record.value());
            log.info("收到一条交通流量数据，采集时间："+jsonObject.getString("tjsj"));
            ack.acknowledge();
        }
    }

}